package com.lmind.core.publish;

import java.io.IOException;
import java.io.Serializable;
import java.net.InetSocketAddress;
import java.util.concurrent.ScheduledExecutorService;

import org.apache.mina.core.service.AbstractIoAcceptor;
import org.apache.mina.core.service.IoService;
import org.apache.mina.transport.socket.nio.NioSocketConnector;

import com.lmind.core.async.ResultHandler;
import com.lmind.core.message.AsyncMessageProcessor;
import com.lmind.core.message.DuplexMessagePipe;
import com.lmind.core.message.DuplexMessageServiceListener;
import com.lmind.core.message.JavaMessageCodec;
import com.lmind.core.message.MessagePipe;
import com.lmind.core.message.MinaDuplexMessageService;

/**
 * 
 * @author 刘勉
 *
 */
public class SocketPublisherRouter<P> extends AbstractPublisher<P> implements Publisher<P> {

	private MinaDuplexMessageService service;
	
	private ScheduledExecutorService scheduledExecutor;
	
	private SimplePublisherRouter<P> router = new SimplePublisherRouter<P>();
	
	private volatile MessagePipe clientPipe;
	
	private DuplexMessageServiceListener connectionListener = new DuplexMessageServiceListener() {
		public void disconnected(DuplexMessagePipe pipe) {
			// 连接断开后应该清理该连接所有的订阅信息
			PipeSubscriber ps = (PipeSubscriber)pipe.getAttribute(PipeSubscriber.class.getName());
			router.cleanSubscribe(ps);
		}
		public void connected(DuplexMessagePipe pipe) {
			clientPipe = pipe;
			// 为每个连接只创建唯一一个PipeSubscriber
			pipe.setAttribute(PipeSubscriber.class.getName(), new PipeSubscriber(pipe));
		}
	}; 
	
	public void setPublisher(Publisher<P> publisher) {
		router.setPublisher(publisher);
	}

	public void setScheduledExecutor(ScheduledExecutorService scheduledExecutor) {
		this.scheduledExecutor = scheduledExecutor;
	}
	
	public void bindIoService(IoService ioService) {
		if (service != null)
			throw new IllegalStateException("已绑定IoService");
		
		service = new MinaDuplexMessageService();
		service.setCodec(new JavaMessageCodec());
		service.setScheduledExecutor(scheduledExecutor);
		service.setMessageProcessor(new ServerProcessor());
		service.addDuplexMessageServiceListener(connectionListener);
		service.bindIoService(ioService);
	}

	public void bind(AbstractIoAcceptor acceptor, InetSocketAddress address) throws IOException {
		if (service != null)
			throw new IllegalStateException("服务已经启动");

		service = new MinaDuplexMessageService();
		service.setCodec(new JavaMessageCodec());
		service.setScheduledExecutor(scheduledExecutor);
		service.setMessageProcessor(new ServerProcessor());
		service.addDuplexMessageServiceListener(connectionListener);
		service.bindIoService(acceptor);
		
		acceptor.bind(address);
	}
	
	public void connect(InetSocketAddress address) {
		if (service != null)
			throw new IllegalStateException("服务已经启动");
		
		NioSocketConnector c = new NioSocketConnector();
		
		service = new MinaDuplexMessageService();
		service.setCodec(new JavaMessageCodec());
		service.setScheduledExecutor(scheduledExecutor);
		service.setMessageProcessor(new ClientProcessor());
		service.addDuplexMessageServiceListener(connectionListener);
		service.bindIoService(c);
		
		c.connect(address);
	}
	
	public void close() {
		if (service != null)
			service.close();
	}
	
	
	@Override
	public void subscribe(Subscriber<P> subscriber, String name)
			throws SubscribeException {
		doSubscribe(subscriber, name);
		SubscribeMessage msg = new SubscribeMessage();
		msg.name = name;
		clientPipe.send(msg);
	}

	@Override
	public void unsubscribe(Subscriber<P> subscriber, String name)
			throws SubscribeException {
		doUnsubscribe(subscriber, name);
	}

	@Override
	public void cleanSubscribe(Subscriber<P> subscriber)
			throws SubscribeException {
		doCleanSubscribe(subscriber);
	}
	
	public static class Message implements Serializable {
	}
	public static class SubscribeMessage extends Message {
		private String name;
	}
	public static class UnsubscribeMessage extends Message {
		private String name;
	}
	public static class PublishMessage<P> extends Message {
		private String name;
		private P pub;
	}
	
	/**
	 * 为每个连接只创建一个PipeSubscriber
	 * @author 刘勉
	 *
	 */
	class PipeSubscriber implements Subscriber<P> {
		private MessagePipe pipe;
		public PipeSubscriber(MessagePipe pipe) {
			this.pipe = pipe;
		}
		@Override
		public void feed(String name, P pub) {
			PublishMessage<P> msg = new PublishMessage<P>();
			msg.name = name;
			msg.pub = pub;
			pipe.send(msg);
		}
	}
	
	
	class ServerProcessor implements AsyncMessageProcessor<DuplexMessagePipe, Object> {
		@Override
		public void process(DuplexMessagePipe context, Object message,
				ResultHandler<Object> handler) {
			
			if (message instanceof SubscribeMessage) {
				SubscribeMessage msg = (SubscribeMessage)message;
				PipeSubscriber ps = (PipeSubscriber)context.getAttribute(PipeSubscriber.class.getName());
				router.subscribe(ps, msg.name);
			} else if (message instanceof UnsubscribeMessage) {
				
			}
			
			if (handler != null)
				handler.result(new Message());
		}
	}
	
	class ClientProcessor implements AsyncMessageProcessor<DuplexMessagePipe, Object> {
		@Override
		public void process(DuplexMessagePipe context, Object message,
				ResultHandler<Object> handler) {
			
			if (message instanceof PublishMessage) {
				PublishMessage<P> msg = (PublishMessage<P>)message;
				doPublish(msg.name, msg.pub);
			}
			
			if (handler != null)
				handler.result(new Message());
			
		}
	}

}
